package org.example.work1.order.rocketmq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.example.work1.order.entity.Account;
import org.example.work1.order.redis.RedisCache;
import org.example.work1.order.service.AccountService;
import org.example.work1.order.service.OrderService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;

/**
 * @ClassName: ConsumerServiceImpl
 * @Description: 消费端执行业务
 * @Author: yanrong
 * @Date: 2019/9/10 17:31
 * @Version: 1.0
 */
@Service
@Slf4j
public class ConsumerServiceImpl implements ConsumerService {
    @Autowired
    OrderService orderService;

    @Autowired
    AccountService accountService;

    @Autowired
    RedisCache redisCache;

    //订单消息消费
    @Transactional
    @Override
    public synchronized ConsumeConcurrentlyStatus consumeAccountMessage(List<MessageExt> messages) {
        try {
            for (MessageExt msg : messages) {
                JSONObject jsonInfo = JSON.parseObject(new String(msg.getBody()));
                if (jsonInfo == null ) {
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                //通过记录表查看消息是否已经被消费。
                boolean msgStatus = this.checkTransferStatus(msg.getTransactionId());
                if (msgStatus) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                Account account = JSON.parseObject(jsonInfo.toString(), Account.class);
                account.setTransactionId(msg.getTransactionId());
                if (accountService.saveOrUpdate(account)){
                    log.info("消息消费成功：" + System.currentTimeMillis());
                    //写入redis 缓存  redisCache.set(key,timeout,account) redis集群使用分布式锁

                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } else {
                    //重试
                    log.info("消息消费失败：" + System.currentTimeMillis());
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        } catch (Exception e) {
            throw new RuntimeException("消费端发生异常", e);
        }
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }


    /**
     * 检查本地事务执行状态
     *
     * @param transactionId
     * @return
     */
    public boolean checkTransferStatus(String transactionId) {
        //根据本地线程Id是否有这个数据，有运单记录，标识本地事务执行成功
        try {
            int count = accountService.count(Wrappers.<Account>lambdaQuery().eq(Account::getTransactionId, transactionId));
            return count > 0;
        } catch (Exception e) {
         throw new RuntimeException("消息检查回调失败");
        }
    }

}
